package com.markhsiu.minimq.remote.transport.talentaio;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteOrder;
import java.util.concurrent.TimeUnit;

import com.markhsiu.minimq.client.AbstractClient;
import com.markhsiu.minimq.client.CallBackCache;
import com.markhsiu.minimq.core.exeption.MiniMQException;
import com.markhsiu.minimq.message.Message;
import com.markhsiu.minimq.message.Result;
import com.markhsiu.minimq.remote.Address;
import com.markhsiu.minimq.remote.transport.talentaio.handler.MessageClientAioHandler;
import com.markhsiu.minimq.serialize.SerializeCodec;
import com.markhsiu.minimq.serialize.protostuff.ProtostuffCodec;
import com.talent.aio.client.AioClient;
import com.talent.aio.client.ClientChannelContext;
import com.talent.aio.client.ClientGroupContext;
import com.talent.aio.client.intf.ClientAioHandler;
import com.talent.aio.client.intf.ClientAioListener;
import com.talent.aio.common.Aio;
import com.talent.aio.common.Node;
import com.talent.aio.common.ReconnConf;

public class TalentAioClient extends AbstractClient  {
	
	private Node serverNode = null;
	private AioClient<Object, MessagePacket, Object> aioClient;
	private ClientGroupContext<Object, MessagePacket, Object> clientGroupContext = null;
	private ClientAioHandler<Object, MessagePacket, Object> aioClientHandler = null;
	private ClientAioListener<Object, MessagePacket, Object> aioListener = null;
	// 用来自动连接的，不想自动连接请传null
	private ReconnConf<Object, MessagePacket, Object> reconnConf = new ReconnConf<Object, MessagePacket, Object>(5000L);

	private ClientChannelContext<Object, MessagePacket, Object> clientChannelContext;

	public TalentAioClient() {
		super(null);
	}

	public TalentAioClient(Address serverURL) {
		super(serverURL);
	}
	

	@Override
	public Result send(Message message) {
		if(isDebugEnabled){
			logger.debug("{} send",message);
		}
		System.out.println();
		final Result result = new Result();
		String messageID = message.getMessageID();

		try {
			SerializeCodec<Message> codec = ProtostuffCodec.OneInstance();
			byte[] body = codec.serialize(message);
			MessagePacket packet = new MessagePacket();
			packet.setBody(body);
			
			CallBackCache.putCallBack(messageID, result);
			Aio.send(clientChannelContext, packet);

			Message ask = result.getMessageResult(10, TimeUnit.SECONDS);
			if (ask == null) {
				throw new MiniMQException("响应超时");
			}
			result.setMessage(ask);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			CallBackCache.removeCallBack(messageID);
		}

		return result;
	}
	

	

	@Override
	public void init() {
		String serverIp = addr.getHostname();
		int serverPort = addr.getPort();

		serverNode = new Node(serverIp, serverPort);
		aioClientHandler = new MessageClientAioHandler(callBackor);
		aioListener = null;

		clientGroupContext = new ClientGroupContext<>(aioClientHandler, aioListener, reconnConf);
		clientGroupContext.setReadBufferSize(2048);
		clientGroupContext.setByteOrder(ByteOrder.BIG_ENDIAN);
		clientGroupContext.setEncodeCareWithChannelContext(false);

		try {
			aioClient = new AioClient<>(clientGroupContext);
		} catch (IOException e) {
			e.printStackTrace();
		}

	}
	
	
	@Override
	public void connect() {
		try {
			clientChannelContext = aioClient.connect(serverNode);
			active = true;
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void close() {
		active = false;
	}

	

}
